package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.app.func.DorisSinkFunction;
import com.atguigu.edu.realtime.beans.CartAddUuBean;
import com.atguigu.edu.realtime.beans.DwsTradeCourseOrderWindowBean;
import com.atguigu.edu.realtime.beans.DwsTradeSourceOrderWindowBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.DorisUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.xml.crypto.*;
import javax.xml.crypto.dsig.keyinfo.KeyInfo;
import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * 交易域来源粒度下单统计表
 */
public class DwsTradeSourceOrderWindow {
    public static void main(String[] args) {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(5);
        //TODO 2.检查点相关的设置
         env.enableCheckpointing(5000L);
        //TODO 3.从kakfa主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_source_order_window";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
                = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.对流中数据进行类型转换并过滤掉空消息  jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            if (jsonObj.getString("source_id") != null && jsonObj.getString("order_id") != null) {
                                out.collect(jsonObj);
                            }
                        }
                    }
                }
        );

         //jsonObjDS.print(">>>");
        //TODO 5.去重
        //5.1 按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdKeyedDS = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        //5.3 去重方式 : 状态 + 抵消
        SingleOutputStreamOperator<JSONObject> distinctDS = orderDetailIdKeyedDS.process(

                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    private ValueState<JSONObject> lastJsonObjState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<JSONObject> valueStateDescriptor
                                = new ValueStateDescriptor<JSONObject>("lastJsonObjState", JSONObject.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(10)).build());
                        lastJsonObjState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        //判断是否重复
                        JSONObject lastJsonObj = lastJsonObjState.value();

                        if (lastJsonObj != null) {
                            //说明重复了，将状态中影响到度量值的字段进行取反，再发送到下游进行抵消
                            String OriginalAmount = lastJsonObj.getString("origin_amount");
                            String CouponAmount = lastJsonObj.getString("coupon_reduce");
                            String orderTotalAmount = lastJsonObj.getString("final_amount");


                            //取反，发送到下游去
                            lastJsonObj.put("origin_amount", "-" + OriginalAmount);
                            lastJsonObj.put("coupon_reduce", "-" + CouponAmount);
                            lastJsonObj.put("final_amount", "-" +orderTotalAmount);
                            out.collect(lastJsonObj);

                        }
                        lastJsonObjState.update(jsonObj);
                        out.collect(jsonObj);
                    }
                }
        );
        //distinctDS.print("%%%%");
        // TODO 6.按照来源id和用户id分组
        KeyedStream<JSONObject, String> userIdSourceIdDS = distinctDS.keyBy(
                new KeySelector<JSONObject, String>() {
                    @Override
                    public String getKey(JSONObject jsonObj) throws Exception {
                        return jsonObj.getString("user_id") + jsonObj.getString("source_id");
                    }
                }
        );

        // TODO 7.统计独立用户
        SingleOutputStreamOperator<DwsTradeSourceOrderWindowBean> uuBeanStream = userIdSourceIdDS.process(
                new KeyedProcessFunction<String, JSONObject, DwsTradeSourceOrderWindowBean>() {
                    ValueState<String> lastOrderDtState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> lastOrderDtStateDesc = new ValueStateDescriptor<>("last_order_dt_state", String.class);
                        lastOrderDtStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastOrderDtState = getRuntimeContext().getState(lastOrderDtStateDesc);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<DwsTradeSourceOrderWindowBean> out) throws Exception {
                        long ts = jsonObj.getLong("ts") * 1000;
                        String lastOrderDt = lastOrderDtState.value();
                        String curDt = DateFormatUtil.toDate(ts);
                        Long orderUuCount = 0L;
                        if (lastOrderDt == null || lastOrderDt.compareTo(curDt) < 0) {
                            // 判断为独立用户
                            orderUuCount = 1L;
                            lastOrderDtState.update(curDt);
                        }
                        out.collect(DwsTradeSourceOrderWindowBean.builder()
                                .sourceId(jsonObj.getString("source_id"))
                                .orderTotalAmount(new BigDecimal(jsonObj.getString("final_amount")))
                                .orderUuCount(orderUuCount)
                                .orderId(jsonObj.getString("order_id"))
                                .ts(ts)
                                .build());
                    }
                });
        // uuBeanStream.print(">>>");
        // TODO 8.按照订单id分组
        KeyedStream<DwsTradeSourceOrderWindowBean, String> orderIdStream = uuBeanStream.keyBy(
            new KeySelector<DwsTradeSourceOrderWindowBean, String>() {
            @Override
            public String getKey(DwsTradeSourceOrderWindowBean value) throws Exception {
                return value.getOrderId();
            }
        });
        // TODO 9.统计订单数
        SingleOutputStreamOperator<DwsTradeSourceOrderWindowBean> orderBeanStream = orderIdStream.process(
         new KeyedProcessFunction<String, DwsTradeSourceOrderWindowBean, DwsTradeSourceOrderWindowBean>() {

            ValueState<String> lastOrderIdState = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastOrderStateDesc = new ValueStateDescriptor<>("last_order_state", String.class);
                lastOrderStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(1)).build());
                lastOrderIdState = getRuntimeContext().getState(lastOrderStateDesc);
            }

            @Override
            public void processElement(DwsTradeSourceOrderWindowBean bean, Context ctx, Collector<DwsTradeSourceOrderWindowBean> out) throws Exception {
                String lastOrderId = lastOrderIdState.value();
                Long orderCount = 0L;
                String orderId = bean.getOrderId();
                if (lastOrderId == null) {
                    orderCount = 1L;
                    lastOrderIdState.update(orderId);
                }
                bean.setOrderCount(orderCount);
                out.collect(bean);
            }
        });
        // orderBeanStream.print(">>>");
        // TODO 10.设置水位线
        SingleOutputStreamOperator<DwsTradeSourceOrderWindowBean> withWaterMarkStream
                = orderBeanStream.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<DwsTradeSourceOrderWindowBean>forBoundedOutOfOrderness(Duration.ofSeconds(5L))
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<DwsTradeSourceOrderWindowBean>() {
            @Override
            public long extractTimestamp(DwsTradeSourceOrderWindowBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        // TODO 11.按照统计维度source_id进行分组
        KeyedStream<DwsTradeSourceOrderWindowBean, String> sourceIdStream = withWaterMarkStream.keyBy(
            new KeySelector<DwsTradeSourceOrderWindowBean, String>() {
            @Override
            public String getKey(DwsTradeSourceOrderWindowBean value) throws Exception {

                return value.getSourceId();
            }
        });
        //withWaterMarkStream.print(">>>");
        // TODO 12 开窗 聚合计算
        SingleOutputStreamOperator<DwsTradeSourceOrderWindowBean> reduceStream = sourceIdStream.window(
                TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(2L)))
                .reduce(new ReduceFunction<DwsTradeSourceOrderWindowBean>() {
                    @Override
                    public DwsTradeSourceOrderWindowBean reduce(DwsTradeSourceOrderWindowBean value1, DwsTradeSourceOrderWindowBean value2) throws Exception {
                        value1.setOrderTotalAmount(value1.getOrderTotalAmount().add(value2.getOrderTotalAmount()));
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setOrderUuCount(value1.getOrderUuCount() + value2.getOrderUuCount());
//                       System.out.println(value1);
                        return value1;
                    }
                }, new ProcessWindowFunction<DwsTradeSourceOrderWindowBean, DwsTradeSourceOrderWindowBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<DwsTradeSourceOrderWindowBean> elements, Collector<DwsTradeSourceOrderWindowBean> out) throws Exception {

                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String curDate = DateFormatUtil.toDate(context.window().getStart());
                        for (DwsTradeSourceOrderWindowBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setTs(System.currentTimeMillis());
                            element.setCurDate(curDate);
                            out.collect(element);
                        }
                    }
                });
//        reduceStream.print(">>");
        // TODO 14 维度关联来源名称
        SingleOutputStreamOperator<DwsTradeSourceOrderWindowBean> sourceNameStream
                = AsyncDataStream.unorderedWait(reduceStream, new DimAsyncFunction<DwsTradeSourceOrderWindowBean>("dim_base_source") {

            @Override
            public void join(JSONObject jsonObject, DwsTradeSourceOrderWindowBean bean) {
                bean.setSourceName(jsonObject.getString("source_site"));
            }

            @Override
            public String getKey(DwsTradeSourceOrderWindowBean obj) {
                return obj.getSourceId();
            }
        }, 60 , TimeUnit.SECONDS);

        // TODO 15 写出到doris
        sourceNameStream.print("&&&&");
        sourceNameStream.addSink(new DorisSinkFunction<DwsTradeSourceOrderWindowBean>("dws_trade_source_order_window"));
         /*sourceNameStream
                 .map(new BeanToJsonStrFunction<>())
                 .sinkTo(DorisUtil.getDorisSink("dws_trade_source_order_window"));*/

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }


    }
}
